/*
 * iothread.cpp
 *
 *  Created on: 2011-8-22
 *      Author: chenl
 *     Purpose:
 *        Note:
 */

#include "iothread.h"
#include "config.h"
#include "common/logapi.h"
#include "dataqueue.h"
#include "global.h"
#ifndef WIN32
#include <sys/socket.h>
#endif
#include <time.h>
#include <string.h>

enum
{
	enmDefaultThreadExecutePeriod = 10000,
};

CIOThread::CIOThread() : m_nListenFD(0), m_nClientCount(0)
{
	memset(m_arrClientPool, 0, sizeof(m_arrClientPool));
}

CIOThread::~CIOThread()
{

}

CIOThread::Execute()
{
	bool bIdle = false;

	//initialize server
	int ret = m_tcpServer.Start();
	if (0 > ret)
	{
		WRITE_SERVER_LOG(enmLogLevel_FATAL, "tcp server failed to start! server is going down!");
		return ;
	}

	//initialize sensor
	ret = m_sensor.SetSerialPortSetting(GET_CONFIG_INSTANCE().GetAdapter());
	if (0 > ret)
	{
		WRITE_SERVER_LOG(enmLogLevel_FATAL, "failed to connect sensor, set config failed! server is going down!");
		return ;
	}
	ret = m_sensor.Connect();
	if (0 > ret)
	{
		WRITE_SERVER_LOG(enmLogLevel_FATAL, "failed to connect sensor! server is going down!");
		return ;
	}


	struct timeval tv = {0};

	while(!IsTerminated())
	{
		fd_set readSet;
		FD_ZERO(&readSet);

		//new client
		FD_SET(m_tcpServer.GetFD(), &readSet);
		int nReady = select(MAX_FD + 1, &readSet, NULL, NULL, &tv);
		if (0 <= nReady)
		{
			// find an empty client
			for (int i = 0; i < MAX_FD; ++i)
			{
				if (NULL == m_arrClientPool[i])
					break;
			}
			if (i >= MAX_FD)
			{
				i = TryCleanClientPool();
				WRITE_SERVER_LOG(enmLogLevel_WARN, "Too many Clients! Try Cleaning Client pool");
			}
			if (0 < i && MAX_FD > i)
			{
				if (FD_ISSET(m_tcpServer.GetFD(), &readSet))
				{
					int fd = enmInvalidFD;
					m_tcpServer.Accept(fd);
					m_arrClientPool[i] = new CTcpClient(fd);
					bIdle = false;
				}
			}
			else
			{
				WRITE_SERVER_LOG(enmLogLevel_WARN, "Client pool is full!");
			}
		}
		//read package from client
		for (int i = 0; i < MAX_FD; ++i)
		{
			if (NULL == m_arrClientPool[i])
			{
				continue;
			}
			if (CConnect::enmConnStatus_Connected == m_arrClientPool[i]->GetConnectonStatus())
			{
				FD_SET(m_arrClientPool[i]->GetFD(), &readSet);
			}
		}
		FD_SET(m_sensor.GetFD(), &readSet);

		nReady = select(MAX_FD + 1, &readSet, NULL, NULL, &tv);
		for (int i = 0; i < nReady; ++i)
		{
			if (FD_ISSET(m_sensor.GetFD(), &readSet))
			{
				Byte buf[MAX_PACKED_PKG_SIZE] = {0};
				uint16_t uLen = 0;

				uLen = m_sensor.Recv(buf, sizeof(buf));
				if (0 >= uLen)
				{
					WRITE_SERVER_LOG(enmLogLevel_WARN, "Receive empty package from client %d", j);
					continue;
				}
				BuffParam buff = {0};
				buff.pBuff = buf;
				buff.len = uLen;
				GET_RADARDATAQUEUE_INSTANCE().Push(buff);
			}
			for (int j = 0; i < nReady && j < MAX_FD; ++j)
			{
				if (FD_ISSET(m_arrClientPool[j]->GetFD(), &readSet))
				{
					Byte buf[MAX_PACKED_PKG_SIZE] = {0};
					uint16_t uLen = 0;
					uLen = m_arrClientPool[j]->Recv(buf, sizeof(buf));
					if (0 >= uLen)
					{
						WRITE_SERVER_LOG(enmLogLevel_WARN, "Receive empty package from client %d", j);
						continue;
					}
					GET_CLIENTDATAQUEUE_INSTANCE().Push(buf, uLen, j);
				}
			}
			bIdle = false;
		}
		//send package to sensor
		Byte buf[MAX_PACKED_PKG_SIZE] = {0};
		uint16_t uLen = 0;
		ClientID id = 0;
		GET_CLIENTDATAQUEUE_INSTANCE().Pop(buf, uLen, id);
		if (0 < uLen)
		{
			for (uint16_t nSent = 0; uLen > nSent; uLen -= nSent)
			{
				nSent = m_sensor.Send(buf, uLen);
				if (-1 == nSent)
				{
					WRITE_SERVER_LOG(enmLogLevel_ERROR, "failed to send package to sensor %d!error=0x%08X", m_sensor.GetError());
				}
			}
			bIdle = false;
		}

		//send package to client
		Byte buf[MAX_PACKED_PKG_SIZE] = {0};
		uint16_t uLen = sizeof(buf);
		BuffParam buff = {0};
		buff.pBuff = buf;
		buff.len = uLen;
		GET_RADARDATAQUEUE_INSTANCE().Pop(buff);
		if (0 < buff.len)
		{
			for (unsigned int i = 0; i < MAX_FD; ++i)
			{
				for (int nSent = 0; uLen > nSent; uLen -= nSent)
				{
					nSent = m_arrClientPool[i]->Send(buff.pBuff, buff.len);
					if (-1 == nSent)
					{
						WRITE_SERVER_LOG(enmLogLevel_ERROR, "failed to send package to client %d!error=0x%08X", i, m_arrClientPool[i]->GetError());
					}
				}
			}
			bIdle = false;
		}

		//no package
		if (bIdle)
		{
			TryCleanClientPool();
			DELAY(enmDefaultThreadExecutePeriod);
		}
	}
	ret = m_tcpServer.Stop();
	if (0 > ret)
	{
		WRITE_SERVER_LOG(enmLogLevel_FATAL, "tcp server failed to stop! server is going down!");
	}

}

int CIOThread::TryCleanClientPool()
{
	int index = MAX_FD;
	for (int i = 0; i < MAX_FD; ++i)
	{
		if (NULL == m_arrClientPool[i])
		{
			continue;
		}
		if (CConnect::enmConnStatus_Closed == m_arrClientPool[i]->GetConnectonStatus()
				|| enmInvalidFD == m_arrClientPool[i]->GetFD())
		{
			delete m_arrClientPool[i];
			m_arrClientPool[i] = NULL;
			index = i;
		}
	}
}
